#include "echo_connection.h"
#include "event_loop.h"
#include "thread_pool.h"
#include <algorithm>
#include <fcntl.h>
#include <sys/epoll.h>
#include <unistd.h>

namespace {
void setNonBlocking(int fd) {
    int flags = ::fcntl(fd, F_GETFL, 0);
    ::fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}
}

EchoConnection::EchoConnection(EventLoop& loop, int fd, ThreadPool& computePool)
    : EventHandler(loop, fd),
      computePool_(computePool) {
    setNonBlocking(fd);
}

EchoConnection::~EchoConnection() = default;

void EchoConnection::handleEvent(uint32_t events) {
    if (events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) {
        shutdownInLoop();
        return;
    }

    if (events & EPOLLIN) {
        char buf[4096];
        while (true) {
            ssize_t n = ::read(fd(), buf, sizeof(buf));
            if (n > 0) {
                std::lock_guard<std::mutex> lk(mutex_);
                readBuf_.append(buf, static_cast<std::size_t>(n));
            } else if (n == 0) {
                shutdownInLoop();
                return;
            } else {
                if (errno == EAGAIN || errno == EWOULDBLOCK) break;
                shutdownInLoop();
                return;
            }
        }
        submitComputeTask();
    }

    if (events & EPOLLOUT) {
        std::lock_guard<std::mutex> lk(mutex_);
        while (!writeBuf_.empty()) {
            ssize_t n = ::write(fd(), writeBuf_.data(), writeBuf_.size());
            if (n > 0) {
                writeBuf_.erase(0, static_cast<std::size_t>(n));
            } else {
                if (errno == EAGAIN || errno == EWOULDBLOCK) break;
                shutdownInLoop();
                return;
            }
        }
        if (writeBuf_.empty()) {
            loop().updateHandler(fd(), EPOLLIN | EPOLLET);
        }
    }
}

void EchoConnection::submitComputeTask() {
    std::string payload;
    {
        std::lock_guard<std::mutex> lk(mutex_);
        if (readBuf_.empty()) return;
        payload.swap(readBuf_);
    }

    if (closed_) return;

    computePool_.enqueue([this, payload = std::move(payload)]() mutable {
        // === Compute 阶段：模拟耗时处理 ===
        std::reverse(payload.begin(), payload.end());

        // 回到 I/O 线程完成 Encode & Send
        loop().runInLoop([this, payload = std::move(payload)]() mutable {
            onComputeDone(std::move(payload));
        });
    });
}

void EchoConnection::onComputeDone(std::string response) {
    if (closed_) return;

    {
        std::lock_guard<std::mutex> lk(mutex_);
        writeBuf_.append(std::move(response));
    }

    loop().updateHandler(fd(), EPOLLIN | EPOLLOUT | EPOLLET);
}

void EchoConnection::shutdownInLoop() {
    if (closed_) return;
    closed_ = true;
    loop().removeHandler(fd());
    delete this;  // handler 生命周期由事件循环掌握
}